package my.yuan.ms.rediskafkams.producer;

import lombok.extern.slf4j.Slf4j;
import my.yuan.ms.rediskafkams.controller.vo.MsVO;
import my.yuan.ms.rediskafkams.service.StockService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @author zzMh
 * @Description TODO
 */
@Component
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, MsVO> kafkaTemplate;
    @Autowired
    private StockService stockService;

    public static final String GROUP_ID = "group-demo";
    public static final String TOPIC_DEMO = "topic-demo";
    public static final int partition = 0;

    public void send(MsVO msVO) {
        ListenableFuture<SendResult<String, MsVO>> future = kafkaTemplate.send(TOPIC_DEMO, String.valueOf(msVO.getUid()), msVO);
        log.info("生产者开始发送消息");
        future.addCallback(stringMsVOSendResult -> {
            MsVO value = stringMsVOSendResult.getProducerRecord().value();
            log.info("发送成功！ 商品id：{},用户id:{}", value.getCid(), value.getUid());
        }, (throwable -> {
            log.info("发送失败！库存回退~");
            stockService.redisPlusStock(msVO.getCid());
        }));
    }
}
